package com.example.flinkcourse.lesson2.source;

import com.example.flinkcourse.lesson2.model.Event;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.pulsar.source.PulsarSource;
import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
import org.apache.flink.streaming.api.functions.source.Source;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Pulsar 数据源工厂类
 * 用于创建 Pulsar 数据源，支持 Kerberos 认证
 */
public class PulsarSourceFactory {
    private static final Logger LOG = LoggerFactory.getLogger(PulsarSourceFactory.class);

    /**
     * 创建 Pulsar 数据源
     * @return Pulsar 数据源
     */
    public static Source<Event> createPulsarSource() {
        LOG.info("Creating Pulsar source...");
        
        // 配置 Pulsar 连接
        PulsarSourceBuilder<Event> builder = PulsarSource.builder();
        
        // 设置服务地址
        builder.setServiceUrl("pulsar://localhost:6650");
        
        // 设置认证信息
        if (isKerberosEnabled()) {
            LOG.info("Kerberos authentication is enabled");
            builder.setAuthPlugin("org.apache.pulsar.client.impl.auth.AuthenticationToken");
            builder.setAuthParams(getKerberosAuthParams());
        }
        
        // 设置订阅和主题
        builder.setSubscriptionName("flink-subscription");
        builder.setTopics("persistent://public/default/input-topic");
        
        // 设置起始位置
        builder.setStartCursor(StartCursor.latest());
        
        // 设置反序列化器
        builder.setDeserializationSchema(
            PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema())
        );
        
        LOG.info("Pulsar source created successfully");
        return builder.build();
    }
    
    /**
     * 检查是否启用 Kerberos 认证
     * @return 是否启用 Kerberos
     */
    private static boolean isKerberosEnabled() {
        // TODO: 从配置中读取是否启用 Kerberos
        return true;
    }
    
    /**
     * 获取 Kerberos 认证参数
     * @return 认证参数
     */
    private static String getKerberosAuthParams() {
        // TODO: 从配置中读取 Kerberos 认证参数
        return "token:xxx";
    }
} 